package com.csw.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo02CreateDF {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .config("spark.sql.shuffle.partitions", 2)
      .appName("createDF")
      .getOrCreate()

    /**
      * 1、读取json文件
      *
      * json中自带列名，不需要指定列名
      */
    val jsonDF: DataFrame = spark
      .read
      .format("json")
      .load("spark/data/students.json")

    /**
      * 2、读取文本文件格式的数据
      *
      */
    val csvDF: DataFrame = spark
      .read
      .format("csv")
      .option("seq", ",") //默认是逗号分隔
      //按照数据的顺序指定表结构
      .schema("id STRING,name STRING , age INT , gender STRING , clazz STRING")
      .load("spark/data/students.txt")


    //    csvDF.printSchema()//显示表格式
    //    csvDF.show()//显示表数据

    /**
      * 读取数据库构建df
      */
    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "user.student")
      .option("user", "root")
      .option("password", "123456")
      .load()

//    jdbcDF.show()

    /**
      * 保存一个parquet格式的文件
      *
      * parquet是一种带表的压缩格式
      */
    csvDF.write.mode(SaveMode.Overwrite).parquet("spark/data/parquet")

    /**
      * 读取parquet 格式的文件
      */
    val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("spark/data/parquet")

//    parquetDF.printSchema()
//    parquetDF.show()


    /**
      * 保存一个orc格式的数据
      */

    csvDF.write.mode(SaveMode.Overwrite).orc("spark/data/orc")

    /**
      * 读取orc格式的文件
      */
    val orcDF: DataFrame = spark
      .read
      .format("orc")
      .load("spark/data/orc")

//    orcDF.printSchema()
//    orcDF.show()


    /**
      * 基于RDD创建DataFrame
      */
    val sc: SparkContext = spark.sparkContext

    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    val studentsMapRDD: RDD[(String, String, Int, String, String)] = studentsRDD.map(i => {
      val split: Array[String] = i.split(",")

      (split(0), split(1), split(2).toInt, split(3), split(4))
    })

    //导入隐式转换
    import spark.implicits._

    //将rdd转换成DF，指定列名
    val studentsDF: DataFrame = studentsMapRDD.toDF("id", "name", "age", "gender", "clazz")

    studentsDF.printSchema()
    studentsDF.show()
  }
}
